package com.project.mq.mqserver;


import com.project.mq.common.Consumer;
import com.project.mq.common.MqException;
import com.project.mq.mqserver.core.*;
import com.project.mq.mqserver.datacenter.DiskDataCenter;
import com.project.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Getter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 *  通过这个类，来表示虚拟主机。
 *  每个虚拟主机下面都管理着自己的交换机、队列、绑定、消息数据。
 *  同时提供 api 供上层调用。
 *  针对 VirtualHost 这个类，作为业务逻辑的整合者，就需要对于代码中抛出的异常进行处理了。
 */
@Getter
public class VirtualHost {
    private String virtualHostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    //操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    //操作队列的锁对象
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualhostName) {
        this.virtualHostName = virtualhostName;

        //对于 MemoryDataCenter 来说，不需要额外的初始化操作的。只要对象 new 出来就行了。
        // 但是，针对 DiskDataCenter 来说，则需要进行初始化操作，建库建表和初始数据的设定。
        // 先初始化硬盘数据
        diskDataCenter.init();
        // 后初始化内存数据
        //memoryDataCenter.init();
        try {
            // 进行恢复操作
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException |MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    /*public String getVirtualhostName() {
        return virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }*/
    // 创建交换机
    //exchangeDeclare 的语义是, 不存在就创建, 存在则直接返回. 因此不叫做 "exchangeCreate".
    // 先写硬盘，后写内存。写硬盘失败概率更大，如果异常了，也就不写内存了。
    //boolean autoDelete, Map<String, Object> arguments并没有实现,只是和RabitMq原始的api尽可能保持一致
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 真实的 exchangeName 需要拼接上 virtualhostName
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 判定该交换机是否存在,直接从内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                // 2.真正创建交换机, 构造 Exchange 对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3. 把数据写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4. 把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成! exchangeName=" + exchangeName);
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    // 删除交换机
    // 先写硬盘，后写内存。写硬盘失败概率更大，如果异常了，也就不写内存了。
    public boolean exchangeDelete(String exchangeName) {
        // 真实的 exchangeName 需要拼接上 virtualhostName
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 先找到对应的交换机.
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在，无法删除!");
                }
                // 2. 删除硬盘上的交换机数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);

                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName=" + queueName);
                    return true;
                }
                // 2. 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 写硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 4. 写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    // 删除队列
    public boolean queueDelete(String queueName) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 根据 queueName 查询对应的队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在，无法删除!");
                }
                // 2. 删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功! queueName=" + queueName);
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    // 创建绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判定 binding 是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(queueName, exchangeName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName + ", exchangeName=" + exchangeName);
                    }
                    // 2. 校验 bindingKey 是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }
                    // 3. 创建 binding 对象
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取到对应的 exchange 和 queue 对象,如果交换机和对列不存在,那么绑定也是无法创建的
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 对应的队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 对应的交换机不存在! exchangeName=" + exchangeName);
                    }
                    // 5. 如果 exchange 和 queue 都是持久化的，则 binding 也持久化
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 创建绑定成功! exchangeName=" + exchangeName + ", queueName=" + queueName);
                }
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建绑定失败! exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    // 解除绑定
    public boolean queueUnbind(String queueName, String exchangeName) {
        // 真实的 queueName 需要拼接上 virtualhostName
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取到 binding
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new Exception("[VirtualHost] 绑定删除失败! 绑定不存在!  exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 2. 无论绑定是否持久化了，都尝试从硬盘删一下。就算不存在，这个删除也无副作用。
                    diskDataCenter.deleteBinding(binding);

                    // 3. 从内存删除 binding
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 绑定删除成功! exchangeName=" + exchangeName + ", queueName=" + queueName);
                }
            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败! exchangeName=" + exchangeName + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列
    // 发送消息
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机名字。如果是 null，则使用默认交换机
            if (exchangeName == null) {
                exchangeName = "";
            }
            exchangeName = virtualHostName + exchangeName;

            // 2. 检查路由键合法性
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法! routingKey=" + routingKey);
            }

            // 3. 查找到交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
            }

            if (exchange.getType() == ExchangeType.DIRECT) {
                //直接交换机的方式进行转发
                //按照直接交换机的方式来转发消息。以routingKey作为队列的名字，直接把消息写入指定的队列中。
                // 此时，可以无视绑定关系。
                String queueName = virtualHostName + routingKey;
                // 4. 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);

                // 5. 直接发送，不需要 binding，直接根据 routingKey 找到队列对象，进行绑定。
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                }
                 // 6. 直接转发消息
                sendMessage(queue, message);
            } else {
                //fanout和topic的方式进行转发
                // 4. 找到交换机对应的绑定对象
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                // 5. 遍历所有绑定, 进⼊消息转发逻辑.
                for (Map.Entry<String, Binding> entry : bindings.entrySet()) {
                    // 1) 判定队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处咱们就不抛出异常了，可能此处有多个这样的队列。
                        // 希望不要因为一个队列的失败，影响到其他队列的消息的传输。
                        System.out.println("[VirtualHost] 队列不存在! queueName=" + binding.getQueueName());
                        continue;
                        //throw new MqException("[VirtualHost] 队列不存在! queueName=" + binding.getQueueName());

                    }
                    // 2) 构造消息对象，针对每次写入队列，都构造一个唯一的消息对象 id。使同一个消息，在不同队列中也能有不同的消息 id。
                    // 如果两个队列中的消息 id 一样，此时就可能在 messageMap 中只存在一份消息，而在 queueMessageMap 中存在多份消息。
                    // 此时针对消息进行消费操作，就可能出现一个队列消费了之后，把消息从 messageMap 删除了；第二次再从另一个队列消费的时候，就无法从 messageMap 中获取到消息了。
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判定能否转发
                    //如果是fanout，所有绑定的队列都要转发的。
                    // 如果是topic，还需要判定下，bindingKey和routingKey是不是匹配。
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发布失败!");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws Exception {
        //此处发送消息，就是把消息写入到硬盘和内存上。
        // 1. 先写硬盘
        // deliverMode 为 1, 表⽰不持久化; 为 2 表⽰持久化. AMQP 协议规定的.
        int deliverMode = message.getDeliverMode();
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        // 2. 再写内存
        memoryDataCenter.sendMessage(queue, message);
        // 3. 通知消费者去消费消息
        consumerManager.notifyConsume(queue.getName());
    }
    // 订阅消息
    // 添加一个队列的订阅者，当队列收到消息之后，就要把消息推送给对应的订阅者，
    // 如果是多个消费者消费一个队列，将使用轮询的方式进行消费。
    // 参数的 consumerTag 消费者的身份标识,应该在网络通信部分设定。
    // autoAck: 消息被消费完成后，应答的方式。为 true 自动应答，为 false 手动应答。
    // consumer: 是一个回调函数。此处类型设定成函数式接口，这样后续调用 basicConsume 并且传实参的时候，就可以写作 lambda 样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象，把这个对应的队列找到，再把这个 Consumer 对象添加到该队列中。
        queueName = virtualHostName + queueName;
        try {
            // 把 consumer 加到监听线程管理的消费者数组中
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //手动应答
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            //获取消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在! messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName=" + queueName);
            }
            // 删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }

            // 删除内存中的数据
            memoryDataCenter.removeMessage(messageId);
            // 删除待 ack 队列中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功! queueName=" + queueName + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败! queueName=" + queueName + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
